// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#ifndef CEPH_LIBRBD_IO_IMAGE_DISPATCH_SPEC_H
#define CEPH_LIBRBD_IO_IMAGE_DISPATCH_SPEC_H

#include "include/int_types.h"
#include "include/buffer.h"
#include "include/Context.h"
#include "common/zipkin_trace.h"
#include "librbd/io/AioCompletion.h"
#include "librbd/io/Types.h"
#include "librbd/io/ReadResult.h"
#include <boost/variant/variant.hpp>
#include <atomic>

namespace librbd
{

    class ImageCtx;

    namespace io
    {

        struct ImageDispatcherInterface;

        class ImageDispatchSpec
        {
        private:
            // helper to avoid extra heap allocation per object IO
            struct C_Dispatcher : public Context
            {
                ImageDispatchSpec *image_dispatch_spec;

                C_Dispatcher(ImageDispatchSpec *image_dispatch_spec)
                    : image_dispatch_spec(image_dispatch_spec) {}

                void complete(int r) override;
                void finish(int r) override;
            };

        public:
            struct Read
            {
                ReadResult read_result;
                int read_flags;

                Read(ReadResult &&read_result, int read_flags)
                    : read_result(std::move(read_result)), read_flags(read_flags) {}
            };

            struct Discard
            {
                uint32_t discard_granularity_bytes;

                Discard(uint32_t discard_granularity_bytes)
                    : discard_granularity_bytes(discard_granularity_bytes)
                {
                }
            };

            struct Write
            {
                bufferlist bl;

                Write(bufferlist &&bl) : bl(std::move(bl))
                {
                }
            };

            struct WriteSame
            {
                bufferlist bl;

                WriteSame(bufferlist &&bl) : bl(std::move(bl))
                {
                }
            };

            struct CompareAndWrite
            {
                bufferlist cmp_bl;
                bufferlist bl;
                uint64_t *mismatch_offset;

                CompareAndWrite(bufferlist &&cmp_bl, bufferlist &&bl,
                                uint64_t *mismatch_offset)
                    : cmp_bl(std::move(cmp_bl)), bl(std::move(bl)),
                      mismatch_offset(mismatch_offset)
                {
                }
            };

            struct Flush
            {
                FlushSource flush_source;

                Flush(FlushSource flush_source) : flush_source(flush_source)
                {
                }
            };

            struct ListSnaps
            {
                SnapIds snap_ids;
                int list_snaps_flags;
                SnapshotDelta *snapshot_delta;

                ListSnaps(SnapIds &&snap_ids, int list_snaps_flags,
                          SnapshotDelta *snapshot_delta)
                    : snap_ids(std::move(snap_ids)), list_snaps_flags(list_snaps_flags),
                      snapshot_delta(snapshot_delta)
                {
                }
            };

            typedef boost::variant<Read,
                                   Discard,
                                   Write,
                                   WriteSame,
                                   CompareAndWrite,
                                   Flush,
                                   ListSnaps>
                Request;

            C_Dispatcher dispatcher_ctx;

            ImageDispatcherInterface *image_dispatcher;
            ImageDispatchLayer dispatch_layer;
            std::atomic<uint32_t> image_dispatch_flags = 0;
            DispatchResult dispatch_result = DISPATCH_RESULT_INVALID;

            AioCompletion *aio_comp;
            Extents image_extents;
            Request request;
            IOContext io_context;
            int op_flags;
            ZTracer::Trace parent_trace;
            uint64_t tid = 0;

            template <typename ImageCtxT = ImageCtx>
            static ImageDispatchSpec *create_read(
                ImageCtxT &image_ctx, ImageDispatchLayer image_dispatch_layer,
                AioCompletion *aio_comp, Extents &&image_extents,
                ReadResult &&read_result, IOContext io_context, int op_flags,
                int read_flags, const ZTracer::Trace &parent_trace)
            {
                return new ImageDispatchSpec(image_ctx.io_image_dispatcher,
                                             image_dispatch_layer, aio_comp,
                                             std::move(image_extents),
                                             Read{std::move(read_result), read_flags},
                                             io_context, op_flags, parent_trace);
            }

            template <typename ImageCtxT = ImageCtx>
            static ImageDispatchSpec *create_discard(
                ImageCtxT &image_ctx, ImageDispatchLayer image_dispatch_layer,
                AioCompletion *aio_comp, uint64_t off, uint64_t len,
                uint32_t discard_granularity_bytes, IOContext io_context,
                const ZTracer::Trace &parent_trace)
            {
                return new ImageDispatchSpec(image_ctx.io_image_dispatcher,
                                             image_dispatch_layer, aio_comp, {{off, len}},
                                             Discard{discard_granularity_bytes},
                                             io_context, 0, parent_trace);
            }

            template <typename ImageCtxT = ImageCtx>
            static ImageDispatchSpec *create_write(
                ImageCtxT &image_ctx, ImageDispatchLayer image_dispatch_layer,
                AioCompletion *aio_comp, Extents &&image_extents,
                bufferlist &&bl, IOContext io_context, int op_flags,
                const ZTracer::Trace &parent_trace)
            {
                return new ImageDispatchSpec(image_ctx.io_image_dispatcher,
                                             image_dispatch_layer, aio_comp,
                                             std::move(image_extents), Write{std::move(bl)},
                                             io_context, op_flags, parent_trace);
            }

            template <typename ImageCtxT = ImageCtx>
            static ImageDispatchSpec *create_write_same(
                ImageCtxT &image_ctx, ImageDispatchLayer image_dispatch_layer,
                AioCompletion *aio_comp, uint64_t off, uint64_t len,
                bufferlist &&bl, IOContext io_context, int op_flags,
                const ZTracer::Trace &parent_trace)
            {
                return new ImageDispatchSpec(image_ctx.io_image_dispatcher,
                                             image_dispatch_layer, aio_comp,
                                             {{off, len}}, WriteSame{std::move(bl)},
                                             io_context, op_flags, parent_trace);
            }

            template <typename ImageCtxT = ImageCtx>
            static ImageDispatchSpec *create_compare_and_write(
                ImageCtxT &image_ctx, ImageDispatchLayer image_dispatch_layer,
                AioCompletion *aio_comp, Extents &&image_extents,
                bufferlist &&cmp_bl, bufferlist &&bl, uint64_t *mismatch_offset,
                IOContext io_context, int op_flags, const ZTracer::Trace &parent_trace)
            {
                return new ImageDispatchSpec(image_ctx.io_image_dispatcher,
                                             image_dispatch_layer, aio_comp,
                                             std::move(image_extents),
                                             CompareAndWrite{std::move(cmp_bl),
                                                             std::move(bl),
                                                             mismatch_offset},
                                             io_context, op_flags, parent_trace);
            }

            template <typename ImageCtxT = ImageCtx>
            static ImageDispatchSpec *create_flush(
                ImageCtxT &image_ctx, ImageDispatchLayer image_dispatch_layer,
                AioCompletion *aio_comp, FlushSource flush_source,
                const ZTracer::Trace &parent_trace)
            {
                return new ImageDispatchSpec(image_ctx.io_image_dispatcher,
                                             image_dispatch_layer, aio_comp, {},
                                             Flush{flush_source}, {}, 0, parent_trace);
            }

            template <typename ImageCtxT = ImageCtx>
            static ImageDispatchSpec *create_list_snaps(
                ImageCtxT &image_ctx, ImageDispatchLayer image_dispatch_layer,
                AioCompletion *aio_comp, Extents &&image_extents, SnapIds &&snap_ids,
                int list_snaps_flags, SnapshotDelta *snapshot_delta,
                const ZTracer::Trace &parent_trace)
            {
                return new ImageDispatchSpec(image_ctx.io_image_dispatcher,
                                             image_dispatch_layer, aio_comp,
                                             std::move(image_extents),
                                             ListSnaps{std::move(snap_ids),
                                                       list_snaps_flags, snapshot_delta},
                                             {}, 0, parent_trace);
            }

            ~ImageDispatchSpec()
            {
                aio_comp->put();
            }

            void send();
            void fail(int r);

        private:
            struct SendVisitor;
            struct IsWriteOpVisitor;
            struct TokenRequestedVisitor;

            ImageDispatchSpec(ImageDispatcherInterface *image_dispatcher,
                              ImageDispatchLayer image_dispatch_layer,
                              AioCompletion *aio_comp, Extents &&image_extents,
                              Request &&request, IOContext io_context, int op_flags,
                              const ZTracer::Trace &parent_trace)
                : dispatcher_ctx(this), image_dispatcher(image_dispatcher),
                  dispatch_layer(image_dispatch_layer), aio_comp(aio_comp),
                  image_extents(std::move(image_extents)), request(std::move(request)),
                  io_context(io_context), op_flags(op_flags), parent_trace(parent_trace)
            {
                ceph_assert(aio_comp->image_dispatcher_ctx == nullptr);
                aio_comp->image_dispatcher_ctx = &dispatcher_ctx;
                aio_comp->get();
            }
        };

    } // namespace io
} // namespace librbd

#endif // CEPH_LIBRBD_IO_IMAGE_DISPATCH_SPEC_H
